eric/onix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor Policy Enforcer to Policy Checker

Browse Source 
- Renamed the `PolicyEnforcer` interface and related implementations to `PolicyChecker` for clarity and consistency.
- Updated configuration keys in YAML files to reflect the new `checkPolicy` terminology.
- Adjusted related code, tests, and documentation to support the new naming convention and ensure compatibility.
- Enhanced comments and examples for the `checkPolicy` configuration to improve usability.
This commit is contained in:
Ayush Rawat
2026-03-23 04:08:13 +05:30
parent ff4d909b7e
commit 80e7b299f1
29 changed files with 2239 additions and 1229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
pkg/plugin/definition/policyChecker.go Normal file
View File

@@ -0,0 +1,17 @@
package definition
import (
"context"
"github.com/beckn-one/beckn-onix/pkg/model"
)
// PolicyChecker interface for policy checking on incoming messages.
type PolicyChecker interface {
CheckPolicy(ctx *model.StepContext) error
}
// PolicyCheckerProvider interface for creating policy checkers.
type PolicyCheckerProvider interface {
New(ctx context.Context, config map[string]string) (PolicyChecker, func(), error)
}

17
pkg/plugin/definition/policyEnforcer.go
View File

@@ -1,17 +0,0 @@
package definition
import (
"context"
"github.com/beckn-one/beckn-onix/pkg/model"
)
// PolicyEnforcer interface for policy enforcement on incoming messages.
type PolicyEnforcer interface {
Run(ctx *model.StepContext) error
}
// PolicyEnforcerProvider interface for creating policy enforcers.
type PolicyEnforcerProvider interface {
New(ctx context.Context, config map[string]string) (PolicyEnforcer, func(), error)
}

194
pkg/plugin/implementation/opapolicychecker/README.md Normal file
View File

@@ -0,0 +1,194 @@
# OPA Policy Checker Plugin
Validates incoming Beckn messages against network-defined business rules using [Open Policy Agent (OPA)](https://www.openpolicyagent.org/) and the Rego policy language. Non-compliant messages are rejected with a `BadRequest` error code.
## Features
- Evaluates business rules defined in Rego policies
- Supports multiple policy sources: remote URL, local file, directory, or OPA bundle (`.tar.gz`)
- Structured result format: `{"valid": bool, "violations": []string}`
- Fail-closed on empty/undefined query results — misconfigured policies are treated as violations
- Runtime config forwarding: adapter config values are accessible in Rego as `data.config.<key>`
- Action-based enforcement: apply policies only to specific beckn actions (e.g., `confirm`, `search`)
## Configuration
```yaml
checkPolicy:
id: opapolicychecker
config:
type: file
location: ./pkg/plugin/implementation/opapolicychecker/testdata/example.rego
query: "data.policy.result"
actions: "confirm,search"
steps:
- checkPolicy
- addRoute
```
### Configuration Parameters
| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| `type` | string | Yes | - | Policy source type: `url`, `file`, `dir`, or `bundle` |
| `location` | string | Yes | - | Path or URL to the policy source (`.tar.gz` for bundles) |
| `query` | string | Yes | - | Rego query path to evaluate (e.g., `data.policy.result`) |
| `actions` | string | No | *(all)* | Comma-separated beckn actions to enforce |
| `enabled` | string | No | `"true"` | Enable or disable the plugin |
| `debugLogging` | string | No | `"false"` | Enable verbose OPA evaluation logging |
| `refreshIntervalSeconds` | string | No | - | Reload policies every N seconds (0 or omit = disabled) |
| *any other key* | string | No | - | Forwarded to Rego as `data.config.<key>` |
## Policy Hot-Reload
When `refreshIntervalSeconds` is set, a background goroutine periodically re-fetches and recompiles the policy source without restarting the adapter:
- **Atomic swap**: the old evaluator stays fully active until the new one is compiled — no gap in enforcement
- **Non-fatal errors**: if the reload fails (e.g., file temporarily unreachable or parse error), the error is logged and the previous policy stays active
- **Goroutine lifecycle**: the reload loop is tied to the adapter context and stops cleanly on shutdown
```yaml
config:
type: file
location: ./policies/compliance.rego
query: "data.policy.result"
refreshIntervalSeconds: "300" # reload every 5 minutes
```
## How It Works
### Initialization (Load Time)
1. **Load Policy Source**: Fetches `.rego` files from the configured `location` — URL, file, directory, or OPA bundle
2. **Compile Policies**: Compiles all Rego modules into a single optimized `PreparedEvalQuery`
3. **Set Query**: Prepares the OPA query from the configured `query` path (e.g., `data.policy.result`)
### Request Evaluation (Runtime)
1. **Check Action Match**: If `actions` is configured, skip evaluation for non-matching actions
2. **Evaluate OPA Query**: Run the prepared query with the full beckn message as `input`
3. **Handle Result**:
- If the query returns no result (undefined) → **violation** (fail-closed)
- If result is `{"valid": bool, "violations": []string}` → use structured format
- If result is a `set` or `[]string` → each string is a violation
- If result is a `bool` → `false` = violation
- If result is a `string` → non-empty = violation
4. **Reject or Allow**: If violations are found, NACK the request with all violation messages
### Supported Query Output Formats
| Rego Output | Behavior |
|-------------|----------|
| `{"valid": bool, "violations": ["string"]}` | Structured result format (recommended) |
| `set()` / `[]string` | Each string is a violation message |
| `bool` (`true`/`false`) | `false` = denied, `true` = allowed |
| `string` | Non-empty = violation |
| Empty/undefined | **Violation** (fail-closed) — indicates misconfigured query path |
## Example Usage
### Local File
```yaml
checkPolicy:
id: opapolicychecker
config:
type: file
location: ./pkg/plugin/implementation/opapolicychecker/testdata/example.rego
query: "data.policy.result"
```
### Remote URL
```yaml
checkPolicy:
id: opapolicychecker
config:
type: url
location: https://policies.example.com/compliance.rego
query: "data.policy.result"
```
### Local Directory (multiple `.rego` files)
```yaml
checkPolicy:
id: opapolicychecker
config:
type: dir
location: ./policies
query: "data.policy.result"
```
### OPA Bundle (`.tar.gz`)
```yaml
checkPolicy:
id: opapolicychecker
config:
type: bundle
location: https://nfo.example.org/policies/bundle.tar.gz
query: "data.retail.validation.result"
```
## Writing Policies
Policies are written in [Rego](https://www.openpolicyagent.org/docs/latest/policy-language/). The plugin passes the full beckn message body as `input` and any adapter config values as `data.config`:
```rego
package policy
import rego.v1
# Default result: valid with no violations.
default result := {
"valid": true,
"violations": []
}
# Compute the result from collected violations.
result := {
"valid": count(violations) == 0,
"violations": violations
}
# Require provider on confirm
violations contains "confirm: missing provider" if {
input.context.action == "confirm"
not input.message.order.provider
}
# Configurable threshold from adapter config
violations contains "delivery lead time too short" if {
input.context.action == "confirm"
lead := input.message.order.fulfillments[_].start.time.duration
to_number(lead) < to_number(data.config.minDeliveryLeadHours)
}
```
See [`testdata/example.rego`](./testdata/example.rego) for a full working example.
## Relationship with Schema Validator
`opapolicychecker` and `schemav2validator` serve different purposes:
- **Schemav2Validator**: Validates message **structure** against OpenAPI/JSON Schema specs
- **OPA Policy Checker**: Evaluates **business rules** via OPA/Rego policies
Configure them side-by-side in your adapter steps as needed.
## Plugin ID vs Step Name
- **Plugin ID** (used in `id:`): `opapolicychecker` (lowercase, implementation-specific)
- **Step name** (used in `steps:` list and YAML key): `checkPolicy` (camelCase verb)
## Dependencies
- `github.com/open-policy-agent/opa` — OPA Go SDK for policy evaluation and bundle loading
## Known Limitations
- **No bundle signature verification**: When using `type: bundle`, bundle signature verification is skipped. This is planned for a future enhancement.
- **Network-level scoping**: Policies apply to all messages handled by the adapter instance. Per-network policy mapping (by `networkId`) is tracked for follow-up.

32
pkg/plugin/implementation/policyenforcer/benchmark_test.go → pkg/plugin/implementation/opapolicychecker/benchmark_test.go
View File

@@ -5,7 +5,7 @@
//
// Run human-readable report: go test -run TestBenchmarkReport -v -count=1
// Run Go benchmarks: go test -bench=. -benchmem -count=1
package policyenforcer
package opapolicychecker
import (
"context"
@@ -101,12 +101,21 @@ func BenchmarkEvaluate_MostlyInactive(b *testing.B) {
dir := b.TempDir()
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(generateDummyRules(n)), 0644)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
if err != nil {
b.Fatalf("NewEvaluator failed: %v", err)
}
ctx := context.Background()
violations, err := eval.Evaluate(ctx, sampleBecknInput)
if err != nil {
b.Fatalf("correctness check failed: %v", err)
}
if len(violations) != 1 || violations[0] != "real_violation" {
b.Fatalf("expected [real_violation], got %v", violations)
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
_, err := eval.Evaluate(ctx, sampleBecknInput)
@@ -126,12 +135,21 @@ func BenchmarkEvaluate_AllActive(b *testing.B) {
dir := b.TempDir()
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(generateActiveRules(n)), 0644)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
if err != nil {
b.Fatalf("NewEvaluator failed: %v", err)
}
ctx := context.Background()
violations, err := eval.Evaluate(ctx, sampleBecknInput)
if err != nil {
b.Fatalf("correctness check failed: %v", err)
}
if len(violations) != n {
b.Fatalf("expected %d violations, got %d", n, len(violations))
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
_, err := eval.Evaluate(ctx, sampleBecknInput)
@@ -154,7 +172,7 @@ func BenchmarkCompilation(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
_, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
_, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
if err != nil {
b.Fatalf("NewEvaluator failed: %v", err)
}
@@ -188,7 +206,7 @@ func TestBenchmarkReport(t *testing.T) {
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(generateDummyRules(n)), 0644)
start := time.Now()
_, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
_, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
elapsed := time.Since(start)
if err != nil {
t.Fatalf("NewEvaluator(%d rules) failed: %v", n, err)
@@ -209,7 +227,7 @@ func TestBenchmarkReport(t *testing.T) {
dir := t.TempDir()
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(generateDummyRules(n)), 0644)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
if err != nil {
t.Fatalf("NewEvaluator(%d rules) failed: %v", n, err)
}
@@ -245,7 +263,7 @@ func TestBenchmarkReport(t *testing.T) {
dir := t.TempDir()
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(generateActiveRules(n)), 0644)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil, false)
if err != nil {
t.Fatalf("NewEvaluator(%d rules) failed: %v", n, err)
}

14
pkg/plugin/implementation/policyenforcer/cmd/plugin.go → pkg/plugin/implementation/opapolicychecker/cmd/plugin.go
View File

@@ -1,4 +1,4 @@
// Package main provides the plugin entry point for the Policy Enforcer plugin.
// Package main provides the plugin entry point for the OPA Policy Checker plugin.
// This file is compiled as a Go plugin (.so) and loaded by beckn-onix at runtime.
package main
@@ -6,20 +6,20 @@ import (
"context"
"github.com/beckn-one/beckn-onix/pkg/plugin/definition"
"github.com/beckn-one/beckn-onix/pkg/plugin/implementation/policyenforcer"
"github.com/beckn-one/beckn-onix/pkg/plugin/implementation/opapolicychecker"
)
// provider implements the PolicyEnforcerProvider interface for plugin loading.
// provider implements the PolicyCheckerProvider interface for plugin loading.
type provider struct{}
// New creates a new PolicyEnforcer instance.
func (p provider) New(ctx context.Context, cfg map[string]string) (definition.PolicyEnforcer, func(), error) {
enforcer, err := policyenforcer.New(cfg)
// New creates a new PolicyChecker instance.
func (p provider) New(ctx context.Context, cfg map[string]string) (definition.PolicyChecker, func(), error) {
checker, err := opapolicychecker.New(ctx, cfg)
if err != nil {
return nil, nil, err
}
return enforcer, enforcer.Close, nil
return checker, checker.Close, nil
}
// Provider is the exported symbol that beckn-onix plugin manager looks up.

281
pkg/plugin/implementation/opapolicychecker/enforcer.go Normal file
View File

@@ -0,0 +1,281 @@
package opapolicychecker
import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
"sync"
"time"
"github.com/beckn-one/beckn-onix/pkg/log"
"github.com/beckn-one/beckn-onix/pkg/model"
)
// Config holds the configuration for the OPA Policy Checker plugin.
type Config struct {
Type string
Location string
PolicyPaths []string
Query string
Actions []string
Enabled bool
DebugLogging bool
IsBundle bool
RefreshInterval time.Duration // 0 = disabled
RuntimeConfig map[string]string
}
var knownKeys = map[string]bool{
"type": true,
"location": true,
"query": true,
"actions": true,
"enabled": true,
"debugLogging": true,
"refreshIntervalSeconds": true,
}
func DefaultConfig() *Config {
return &Config{
Enabled: true,
RuntimeConfig: make(map[string]string),
}
}
// ParseConfig parses the plugin configuration map into a Config struct.
// Uses type + location pattern (matches schemav2validator).
func ParseConfig(cfg map[string]string) (*Config, error) {
config := DefaultConfig()
typ, hasType := cfg["type"]
if !hasType || typ == "" {
return nil, fmt.Errorf("'type' is required (url, file, dir, or bundle)")
}
config.Type = typ
location, hasLoc := cfg["location"]
if !hasLoc || location == "" {
return nil, fmt.Errorf("'location' is required")
}
config.Location = location
switch typ {
case "url":
for _, u := range strings.Split(location, ",") {
u = strings.TrimSpace(u)
if u != "" {
config.PolicyPaths = append(config.PolicyPaths, u)
}
}
case "file":
config.PolicyPaths = append(config.PolicyPaths, location)
case "dir":
config.PolicyPaths = append(config.PolicyPaths, location)
case "bundle":
config.IsBundle = true
config.PolicyPaths = append(config.PolicyPaths, location)
default:
return nil, fmt.Errorf("unsupported type %q (expected: url, file, dir, or bundle)", typ)
}
query, hasQuery := cfg["query"]
if !hasQuery || query == "" {
return nil, fmt.Errorf("'query' is required (e.g., data.policy.violations)")
}
config.Query = query
if actions, ok := cfg["actions"]; ok && actions != "" {
actionList := strings.Split(actions, ",")
config.Actions = make([]string, 0, len(actionList))
for _, action := range actionList {
action = strings.TrimSpace(action)
if action != "" {
config.Actions = append(config.Actions, action)
}
}
}
if enabled, ok := cfg["enabled"]; ok {
config.Enabled = enabled == "true" || enabled == "1"
}
if debug, ok := cfg["debugLogging"]; ok {
config.DebugLogging = debug == "true" || debug == "1"
}
if ris, ok := cfg["refreshIntervalSeconds"]; ok && ris != "" {
secs, err := strconv.Atoi(ris)
if err != nil || secs < 0 {
return nil, fmt.Errorf("'refreshIntervalSeconds' must be a non-negative integer, got %q", ris)
}
config.RefreshInterval = time.Duration(secs) * time.Second
}
for k, v := range cfg {
if !knownKeys[k] {
config.RuntimeConfig[k] = v
}
}
return config, nil
}
func (c *Config) IsActionEnabled(action string) bool {
if len(c.Actions) == 0 {
return true
}
for _, a := range c.Actions {
if a == action {
return true
}
}
return false
}
// PolicyEnforcer evaluates beckn messages against OPA policies and NACKs non-compliant messages.
type PolicyEnforcer struct {
config *Config
evaluator *Evaluator
evaluatorMu sync.RWMutex
}
// getEvaluator safely returns the current evaluator under a read lock.
func (e *PolicyEnforcer) getEvaluator() *Evaluator {
e.evaluatorMu.RLock()
ev := e.evaluator
e.evaluatorMu.RUnlock()
return ev
}
// setEvaluator safely swaps the evaluator under a write lock.
func (e *PolicyEnforcer) setEvaluator(ev *Evaluator) {
e.evaluatorMu.Lock()
e.evaluator = ev
e.evaluatorMu.Unlock()
}
func New(ctx context.Context, cfg map[string]string) (*PolicyEnforcer, error) {
config, err := ParseConfig(cfg)
if err != nil {
return nil, fmt.Errorf("opapolicychecker: config error: %w", err)
}
evaluator, err := NewEvaluator(config.PolicyPaths, config.Query, config.RuntimeConfig, config.IsBundle)
if err != nil {
return nil, fmt.Errorf("opapolicychecker: failed to initialize OPA evaluator: %w", err)
}
log.Infof(ctx, "OPAPolicyChecker initialized (actions=%v, query=%s, policies=%v, isBundle=%v, debugLogging=%v, refreshInterval=%s)",
config.Actions, config.Query, evaluator.ModuleNames(), config.IsBundle, config.DebugLogging, config.RefreshInterval)
enforcer := &PolicyEnforcer{
config: config,
evaluator: evaluator,
}
if config.RefreshInterval > 0 {
go enforcer.refreshLoop(ctx)
}
return enforcer, nil
}
// refreshLoop periodically reloads and recompiles OPA policies.
// Follows the schemav2validator pattern: driven by context cancellation.
func (e *PolicyEnforcer) refreshLoop(ctx context.Context) {
ticker := time.NewTicker(e.config.RefreshInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
log.Infof(ctx, "OPAPolicyChecker: refresh loop stopped")
return
case <-ticker.C:
e.reloadPolicies(ctx)
}
}
}
// reloadPolicies reloads and recompiles all policies, atomically swapping the evaluator.
// Reload failures are non-fatal; the old evaluator stays active.
func (e *PolicyEnforcer) reloadPolicies(ctx context.Context) {
start := time.Now()
newEvaluator, err := NewEvaluator(
e.config.PolicyPaths,
e.config.Query,
e.config.RuntimeConfig,
e.config.IsBundle,
)
if err != nil {
log.Errorf(ctx, err, "OPAPolicyChecker: policy reload failed (keeping previous policies): %v", err)
return
}
e.setEvaluator(newEvaluator)
log.Infof(ctx, "OPAPolicyChecker: policies reloaded in %s (modules=%v)", time.Since(start), newEvaluator.ModuleNames())
}
// CheckPolicy evaluates the message body against loaded OPA policies.
// Returns a BadReqErr (causing NACK) if violations are found.
// Returns an error on evaluation failure (fail closed).
func (e *PolicyEnforcer) CheckPolicy(ctx *model.StepContext) error {
if !e.config.Enabled {
log.Debug(ctx, "OPAPolicyChecker: plugin disabled, skipping")
return nil
}
action := extractAction(ctx.Request.URL.Path, ctx.Body)
if !e.config.IsActionEnabled(action) {
if e.config.DebugLogging {
log.Debugf(ctx, "OPAPolicyChecker: action %q not in configured actions %v, skipping", action, e.config.Actions)
}
return nil
}
ev := e.getEvaluator()
if e.config.DebugLogging {
log.Debugf(ctx, "OPAPolicyChecker: evaluating policies for action %q (modules=%v)", action, ev.ModuleNames())
}
violations, err := ev.Evaluate(ctx, ctx.Body)
if err != nil {
log.Errorf(ctx, err, "OPAPolicyChecker: policy evaluation failed: %v", err)
return model.NewBadReqErr(fmt.Errorf("policy evaluation error: %w", err))
}
if len(violations) == 0 {
if e.config.DebugLogging {
log.Debugf(ctx, "OPAPolicyChecker: message compliant for action %q", action)
}
return nil
}
msg := fmt.Sprintf("policy violation(s): %s", strings.Join(violations, "; "))
log.Warnf(ctx, "OPAPolicyChecker: %s", msg)
return model.NewBadReqErr(fmt.Errorf("%s", msg))
}
func (e *PolicyEnforcer) Close() {}
func extractAction(urlPath string, body []byte) string {
parts := strings.Split(strings.Trim(urlPath, "/"), "/")
if len(parts) >= 3 {
return parts[len(parts)-1]
}
var payload struct {
Context struct {
Action string `json:"action"`
} `json:"context"`
}
if err := json.Unmarshal(body, &payload); err == nil && payload.Context.Action != "" {
return payload.Context.Action
}
return ""
}

1098
pkg/plugin/implementation/opapolicychecker/enforcer_test.go Normal file
View File

File diff suppressed because it is too large Load Diff

395
pkg/plugin/implementation/opapolicychecker/evaluator.go Normal file
View File

@@ -0,0 +1,395 @@
package opapolicychecker
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"os"
"path"
"path/filepath"
"strings"
"time"
"github.com/open-policy-agent/opa/v1/ast"
"github.com/open-policy-agent/opa/v1/bundle"
"github.com/open-policy-agent/opa/v1/rego"
"github.com/open-policy-agent/opa/v1/storage/inmem"
)
// Evaluator wraps the OPA engine: loads and compiles .rego files at startup,
// then evaluates messages against the compiled policy set.
type Evaluator struct {
preparedQuery rego.PreparedEvalQuery
query string
runtimeConfig map[string]string
moduleNames []string // names of loaded .rego modules
failOnUndefined bool // if true, empty/undefined results are treated as violations
}
// ModuleNames returns the names of the loaded .rego policy modules.
func (e *Evaluator) ModuleNames() []string {
return e.moduleNames
}
// policyFetchTimeout is the HTTP timeout for fetching remote .rego files.
const policyFetchTimeout = 30 * time.Second
// maxPolicySize is the maximum size of a single .rego file fetched from a URL (1 MB).
const maxPolicySize = 1 << 20
// maxBundleSize is the maximum size of a bundle archive (10 MB).
const maxBundleSize = 10 << 20
// NewEvaluator creates an Evaluator by loading .rego files from local paths
// and/or URLs, then compiling them. runtimeConfig is passed to Rego as data.config.
// When isBundle is true, the first policyPath is treated as a URL to an OPA bundle (.tar.gz).
func NewEvaluator(policyPaths []string, query string, runtimeConfig map[string]string, isBundle bool) (*Evaluator, error) {
if isBundle {
return newBundleEvaluator(policyPaths, query, runtimeConfig)
}
return newRegoEvaluator(policyPaths, query, runtimeConfig)
}
// newRegoEvaluator loads raw .rego files from local paths and/or URLs.
func newRegoEvaluator(policyPaths []string, query string, runtimeConfig map[string]string) (*Evaluator, error) {
modules := make(map[string]string)
// Load from policyPaths (resolved locations based on config Type)
for _, source := range policyPaths {
if isURL(source) {
name, content, err := fetchPolicy(source)
if err != nil {
return nil, fmt.Errorf("failed to fetch policy from %s: %w", source, err)
}
modules[name] = content
} else if info, err := os.Stat(source); err == nil && info.IsDir() {
// Directory — load all .rego files inside
entries, err := os.ReadDir(source)
if err != nil {
return nil, fmt.Errorf("failed to read policy directory %s: %w", source, err)
}
for _, entry := range entries {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".rego") || strings.HasSuffix(entry.Name(), "_test.rego") {
continue
}
fpath := filepath.Join(source, entry.Name())
data, err := os.ReadFile(fpath)
if err != nil {
return nil, fmt.Errorf("failed to read policy file %s: %w", fpath, err)
}
modules[entry.Name()] = string(data)
}
} else {
// Local file path
data, err := os.ReadFile(source)
if err != nil {
return nil, fmt.Errorf("failed to read policy file %s: %w", source, err)
}
modules[filepath.Base(source)] = string(data)
}
}
if len(modules) == 0 {
return nil, fmt.Errorf("no .rego policy files found from any configured source")
}
return compileAndPrepare(modules, nil, query, runtimeConfig, true)
}
// newBundleEvaluator loads an OPA bundle (.tar.gz) from a URL and compiles it.
func newBundleEvaluator(policyPaths []string, query string, runtimeConfig map[string]string) (*Evaluator, error) {
if len(policyPaths) == 0 {
return nil, fmt.Errorf("bundle source URL is required")
}
bundleURL := policyPaths[0]
modules, bundleData, err := loadBundle(bundleURL)
if err != nil {
return nil, fmt.Errorf("failed to load bundle from %s: %w", bundleURL, err)
}
if len(modules) == 0 {
return nil, fmt.Errorf("no .rego policy modules found in bundle from %s", bundleURL)
}
return compileAndPrepare(modules, bundleData, query, runtimeConfig, true)
}
// loadBundle downloads a .tar.gz OPA bundle from a URL, parses it using OPA's
// bundle reader, and returns the modules and data from the bundle.
func loadBundle(bundleURL string) (map[string]string, map[string]interface{}, error) {
data, err := fetchBundleArchive(bundleURL)
if err != nil {
return nil, nil, err
}
return parseBundleArchive(data)
}
// fetchBundleArchive downloads a bundle .tar.gz from a URL.
func fetchBundleArchive(rawURL string) ([]byte, error) {
parsed, err := url.Parse(rawURL)
if err != nil {
return nil, fmt.Errorf("invalid URL: %w", err)
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return nil, fmt.Errorf("unsupported URL scheme %q (only http and https are supported)", parsed.Scheme)
}
client := &http.Client{Timeout: policyFetchTimeout}
resp, err := client.Get(rawURL)
if err != nil {
return nil, fmt.Errorf("HTTP request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("HTTP %d from %s", resp.StatusCode, rawURL)
}
limited := io.LimitReader(resp.Body, int64(maxBundleSize)+1)
body, err := io.ReadAll(limited)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if len(body) > maxBundleSize {
return nil, fmt.Errorf("bundle exceeds maximum size of %d bytes", maxBundleSize)
}
return body, nil
}
// parseBundleArchive parses a .tar.gz OPA bundle archive and extracts
// rego modules and data. Signature verification is skipped.
func parseBundleArchive(data []byte) (map[string]string, map[string]interface{}, error) {
loader := bundle.NewTarballLoaderWithBaseURL(bytes.NewReader(data), "")
reader := bundle.NewCustomReader(loader).
WithSkipBundleVerification(true).
WithRegoVersion(ast.RegoV1)
b, err := reader.Read()
if err != nil {
return nil, nil, fmt.Errorf("failed to read bundle: %w", err)
}
modules := make(map[string]string, len(b.Modules))
for _, m := range b.Modules {
modules[m.Path] = string(m.Raw)
}
return modules, b.Data, nil
}
// compileAndPrepare compiles rego modules and prepares the OPA query for evaluation.
func compileAndPrepare(modules map[string]string, bundleData map[string]interface{}, query string, runtimeConfig map[string]string, failOnUndefined bool) (*Evaluator, error) {
// Compile modules to catch syntax errors early
compiler, err := ast.CompileModulesWithOpt(modules, ast.CompileOpts{ParserOptions: ast.ParserOptions{RegoVersion: ast.RegoV1}})
if err != nil {
return nil, fmt.Errorf("failed to compile rego modules: %w", err)
}
// Build store data: merge bundle data with runtime config
store := make(map[string]interface{})
for k, v := range bundleData {
store[k] = v
}
store["config"] = toInterfaceMap(runtimeConfig)
pq, err := rego.New(
rego.Query(query),
rego.Compiler(compiler),
rego.Store(inmem.NewFromObject(store)),
).PrepareForEval(context.Background())
if err != nil {
return nil, fmt.Errorf("failed to prepare rego query %q: %w", query, err)
}
names := make([]string, 0, len(modules))
for name := range modules {
names = append(names, name)
}
return &Evaluator{
preparedQuery: pq,
query: query,
runtimeConfig: runtimeConfig,
moduleNames: names,
failOnUndefined: failOnUndefined,
}, nil
}
// isURL checks if a source string looks like a remote URL.
func isURL(source string) bool {
return strings.HasPrefix(source, "http://") || strings.HasPrefix(source, "https://")
}
// fetchPolicy downloads a .rego file from a URL and returns (filename, content, error).
func fetchPolicy(rawURL string) (string, string, error) {
parsed, err := url.Parse(rawURL)
if err != nil {
return "", "", fmt.Errorf("invalid URL: %w", err)
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return "", "", fmt.Errorf("unsupported URL scheme %q (only http and https are supported)", parsed.Scheme)
}
client := &http.Client{Timeout: policyFetchTimeout}
resp, err := client.Get(rawURL)
if err != nil {
return "", "", fmt.Errorf("HTTP request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", "", fmt.Errorf("HTTP %d from %s", resp.StatusCode, rawURL)
}
// Read with size limit
limited := io.LimitReader(resp.Body, maxPolicySize+1)
body, err := io.ReadAll(limited)
if err != nil {
return "", "", fmt.Errorf("failed to read response body: %w", err)
}
if len(body) > maxPolicySize {
return "", "", fmt.Errorf("policy file exceeds maximum size of %d bytes", maxPolicySize)
}
// Derive filename from URL path
name := path.Base(parsed.Path)
if name == "" || name == "." || name == "/" {
name = "policy.rego"
}
if !strings.HasSuffix(name, ".rego") {
name += ".rego"
}
return name, string(body), nil
}
// Evaluate runs the compiled policy against a JSON message body.
// Returns a list of violation strings (empty = compliant).
func (e *Evaluator) Evaluate(ctx context.Context, body []byte) ([]string, error) {
var input interface{}
if err := json.Unmarshal(body, &input); err != nil {
return nil, fmt.Errorf("failed to parse message body as JSON: %w", err)
}
rs, err := e.preparedQuery.Eval(ctx, rego.EvalInput(input))
if err != nil {
return nil, fmt.Errorf("rego evaluation failed: %w", err)
}
// Fail-closed for bundles: if the query returned no result, the policy_query_path
// is likely misconfigured or the rule doesn't exist in the bundle.
if e.failOnUndefined && len(rs) == 0 {
return []string{fmt.Sprintf("policy query %q returned no result (undefined)", e.query)}, nil
}
return extractViolations(rs)
}
// extractViolations pulls violations from the OPA result set.
// Supported query output formats:
// - map with {"valid": bool, "violations": []string}: structured policy_query_path result
// - []string / set of strings: each string is a violation message
// - bool: false = denied ("policy denied the request"), true = allowed
// - string: non-empty = violation message
// - empty/undefined: allowed (no violations)
func extractViolations(rs rego.ResultSet) ([]string, error) {
if len(rs) == 0 {
return nil, nil
}
var violations []string
for _, result := range rs {
for _, expr := range result.Expressions {
switch v := expr.Value.(type) {
case bool:
// allow/deny pattern: false = denied
if !v {
violations = append(violations, "policy denied the request")
}
case string:
// single violation string
if v != "" {
violations = append(violations, v)
}
case []interface{}:
// Result is a list (from set)
for _, item := range v {
if s, ok := item.(string); ok {
violations = append(violations, s)
}
}
case map[string]interface{}:
// Check for structured result: {"valid": bool, "violations": [...]}
if vs := extractStructuredViolations(v); vs != nil {
violations = append(violations, vs...)
} else {
// Fallback: OPA sometimes returns sets as maps with string keys
for key := range v {
violations = append(violations, key)
}
}
}
}
}
return violations, nil
}
// extractStructuredViolations handles the policy_query_path result format:
// {"valid": bool, "violations": []string}
// Returns the violation strings if the map matches this format, or nil if it doesn't.
func extractStructuredViolations(m map[string]interface{}) []string {
validRaw, hasValid := m["valid"]
violationsRaw, hasViolations := m["violations"]
if !hasValid || !hasViolations {
return nil
}
valid, ok := validRaw.(bool)
if !ok {
return nil
}
violationsList, ok := violationsRaw.([]interface{})
if !ok {
return nil
}
// If valid is true and violations is empty, no violations
if valid && len(violationsList) == 0 {
return []string{}
}
var violations []string
for _, item := range violationsList {
if s, ok := item.(string); ok {
violations = append(violations, s)
}
}
// If valid is false but violations is empty, report a generic violation
if !valid && len(violations) == 0 {
violations = append(violations, "policy denied the request")
}
return violations
}
// toInterfaceMap converts map[string]string to map[string]interface{} for OPA store.
func toInterfaceMap(m map[string]string) map[string]interface{} {
result := make(map[string]interface{}, len(m))
for k, v := range m {
result[k] = v
}
return result
}

52
pkg/plugin/implementation/opapolicychecker/testdata/example.rego vendored Normal file
View File

@@ -0,0 +1,52 @@
package policy
import rego.v1
# Example policy: validation rules for beckn messages.
# This demonstrates the structured result format used with policy_query_path.
#
# Available inputs:
# - input: the full JSON message body (beckn request)
# - data.config: runtime config from the adapter YAML (e.g., minDeliveryLeadHours)
# Default result: valid with no violations.
default result := {
"valid": true,
"violations": []
}
# Compute the result from collected violations.
result := {
"valid": count(violations) == 0,
"violations": violations
}
# Require provider details on confirm
violations contains "confirm: missing provider in order" if {
input.context.action == "confirm"
not input.message.order.provider
}
# Require at least one fulfillment on confirm
violations contains "confirm: order has no fulfillments" if {
input.context.action == "confirm"
not input.message.order.fulfillments
}
# Require billing details on confirm
violations contains "confirm: missing billing info" if {
input.context.action == "confirm"
not input.message.order.billing
}
# Require payment details on confirm
violations contains "confirm: missing payment info" if {
input.context.action == "confirm"
not input.message.order.payment
}
# Require search intent
violations contains "search: missing intent" if {
input.context.action == "search"
not input.message.intent
}

113
pkg/plugin/implementation/policyenforcer/README.md
View File

@@ -1,113 +0,0 @@
# Policy Enforcer Plugin
OPA/Rego-based policy enforcement for beckn-onix adapters. Evaluates incoming beckn messages against configurable policies and NACKs non-compliant requests.
## Overview
The `policyenforcer` plugin is a **Step plugin** that:
- Loads `.rego` policy files from URLs, local directories, or local files
- Evaluates incoming messages against compiled OPA policies
- Returns a `BadReqErr` (NACK) when policy violations are detected
- Fails closed on evaluation errors (treats as NACK)
- Is strictly **opt-in** — adapters that don't reference it are unaffected
## Configuration
All config keys are passed via `map[string]string` in the adapter YAML config.
| Key | Required | Default | Description |
|-----|----------|---------|-------------|
| `policyPaths` | Yes (at least one source required) | `./policies` (if dir exists) | Comma-separated list of policy sources — each entry is auto-detected as a **URL**, **directory**, or **file** |
| `query` | No | `data.policy.violations` | Rego query returning violation strings |
| `actions` | No | *(empty — all actions)* | Comma-separated beckn actions to enforce. When omitted, all actions are evaluated and the Rego policy itself decides which to gate. |
| `enabled` | No | `true` | Enable/disable the plugin |
| `debugLogging` | No | `false` | Enable verbose logging |
| *any other key* | No | — | Forwarded to Rego as `data.config.<key>` |
### Policy Sources
`policyPaths` is the single configuration key for all policy sources. Each comma-separated entry is **auto-detected** as:
- **Remote URL** (`http://` or `https://`): fetched via HTTP at startup
- **Local directory**: all `.rego` files loaded (`_test.rego` excluded)
- **Local file**: loaded directly
```yaml
# Single directory
config:
policyPaths: "./policies"
# Single remote URL
config:
policyPaths: "https://policies.example.com/compliance.rego"
# Mix of URLs, directories, and files
config:
policyPaths: "https://policies.example.com/compliance.rego,./policies,/local/safety.rego"
```
When specifying many sources, use the YAML folded scalar (`>-`) to keep the config readable:
```yaml
config:
policyPaths: >-
https://policies.example.com/compliance.rego,
https://policies.example.com/safety.rego,
./policies,
/local/overrides/rate-limit.rego
```
The `>-` folds newlines into spaces, so the value is parsed as a single comma-separated string.
### Minimal Config
By default, the plugin loads `.rego` files from `./policies` and uses the query `data.policy.violations`. A zero-config setup works if your policies are in the default directory:
```yaml
policyEnforcer:
id: policyenforcer
config: {}
```
Or specify a custom policy location:
```yaml
policyEnforcer:
id: policyenforcer
config:
policyPaths: "./policies/compliance.rego"
```
### Air-Gapped Deployments
For environments without internet access, use local file paths or volume mounts:
```yaml
config:
policyPaths: "/mounted-policies/compliance.rego,/mounted-policies/safety.rego"
```
## Example Config
```yaml
plugins:
policyEnforcer:
id: policyenforcer
config:
policyPaths: >-
/local/policies/,
https://policies.example.com/compliance.rego
minDeliveryLeadHours: "4"
debugLogging: "true"
steps:
- policyEnforcer
- addRoute
```
## Relationship with Schema Validator
`policyenforcer` and `schemavalidator`/`schemav2validator` are **separate plugins** with different responsibilities:
- **Schema Validator**: Validates message **structure** against OpenAPI/JSON Schema specs
- **Policy Enforcer**: Evaluates **business rules** via OPA/Rego policies
They use different plugin interfaces (`SchemaValidator` vs `Step`), different engines, and different error types. Configure them side-by-side in your adapter config as needed.

126
pkg/plugin/implementation/policyenforcer/config.go
View File

@@ -1,126 +0,0 @@
package policyenforcer
import (
"fmt"
"os"
"strings"
)
// Config holds the configuration for the Policy Enforcer plugin.
type Config struct {
// PolicyPaths is a list of policy sources. Each entry is auto-detected as:
// - Remote URL (http:// or https://) → fetched via HTTP
// - Local directory → all .rego files loaded (excluding _test.rego)
// - Local file → loaded directly
// Parsed from the comma-separated "policyPaths" config key.
PolicyPaths []string
// Query is the Rego query that returns a set of violation strings.
// Default: "data.policy.violations".
Query string
// Actions is the list of beckn actions to enforce policies on.
// When empty or nil, all actions are considered and the Rego policy
// is responsible for deciding which actions to gate.
Actions []string
// Enabled controls whether the plugin is active.
Enabled bool
// DebugLogging enables verbose logging.
DebugLogging bool
// RuntimeConfig holds arbitrary key-value pairs passed to Rego as data.config.
// Keys like minDeliveryLeadHours are forwarded here.
RuntimeConfig map[string]string
}
// Known config keys that are handled directly (not forwarded to RuntimeConfig).
var knownKeys = map[string]bool{
"policyPaths": true,
"query": true,
"actions": true,
"enabled": true,
"debugLogging": true,
}
// DefaultConfig returns a Config with sensible defaults.
func DefaultConfig() *Config {
return &Config{
Query: "data.policy.violations",
Enabled: true,
DebugLogging: false,
RuntimeConfig: make(map[string]string),
}
}
// ParseConfig parses the plugin configuration map into a Config struct.
func ParseConfig(cfg map[string]string) (*Config, error) {
config := DefaultConfig()
// Comma-separated policyPaths (each entry auto-detected as URL, directory, or file)
if paths, ok := cfg["policyPaths"]; ok && paths != "" {
for _, p := range strings.Split(paths, ",") {
p = strings.TrimSpace(p)
if p != "" {
config.PolicyPaths = append(config.PolicyPaths, p)
}
}
}
if len(config.PolicyPaths) == 0 {
// Fall back to the default ./policies directory if it exists on disk.
if info, err := os.Stat("./policies"); err == nil && info.IsDir() {
config.PolicyPaths = append(config.PolicyPaths, "./policies")
} else {
return nil, fmt.Errorf("at least one policy source is required (policyPaths)")
}
}
if query, ok := cfg["query"]; ok && query != "" {
config.Query = query
}
if actions, ok := cfg["actions"]; ok && actions != "" {
actionList := strings.Split(actions, ",")
config.Actions = make([]string, 0, len(actionList))
for _, action := range actionList {
action = strings.TrimSpace(action)
if action != "" {
config.Actions = append(config.Actions, action)
}
}
}
if enabled, ok := cfg["enabled"]; ok {
config.Enabled = enabled == "true" || enabled == "1"
}
if debug, ok := cfg["debugLogging"]; ok {
config.DebugLogging = debug == "true" || debug == "1"
}
// Forward unknown keys to RuntimeConfig (e.g., minDeliveryLeadHours)
for k, v := range cfg {
if !knownKeys[k] {
config.RuntimeConfig[k] = v
}
}
return config, nil
}
// IsActionEnabled checks if the given action is in the configured actions list.
// When the actions list is empty/nil, all actions are enabled and action-gating
// is delegated entirely to the Rego policy.
func (c *Config) IsActionEnabled(action string) bool {
if len(c.Actions) == 0 {
return true
}
for _, a := range c.Actions {
if a == action {
return true
}
}
return false
}

106
pkg/plugin/implementation/policyenforcer/enforcer.go
View File

@@ -1,106 +0,0 @@
package policyenforcer
import (
"context"
"encoding/json"
"fmt"
"strings"
"github.com/beckn-one/beckn-onix/pkg/log"
"github.com/beckn-one/beckn-onix/pkg/model"
)
// PolicyEnforcer is a Step plugin that evaluates beckn messages against
// OPA policies and NACKs non-compliant messages.
type PolicyEnforcer struct {
config *Config
evaluator *Evaluator
}
// New creates a new PolicyEnforcer instance.
func New(cfg map[string]string) (*PolicyEnforcer, error) {
config, err := ParseConfig(cfg)
if err != nil {
return nil, fmt.Errorf("policyenforcer: config error: %w", err)
}
evaluator, err := NewEvaluator(config.PolicyPaths, config.Query, config.RuntimeConfig)
if err != nil {
return nil, fmt.Errorf("policyenforcer: failed to initialize OPA evaluator: %w", err)
}
log.Infof(context.TODO(), "PolicyEnforcer initialized (actions=%v, query=%s, policies=%v, debugLogging=%v)",
config.Actions, config.Query, evaluator.ModuleNames(), config.DebugLogging)
return &PolicyEnforcer{
config: config,
evaluator: evaluator,
}, nil
}
// Run implements the Step interface. It evaluates the message body against
// loaded OPA policies. Returns a BadReqErr (causing NACK) if violations are found.
// Returns an error on evaluation failure (fail closed).
func (e *PolicyEnforcer) Run(ctx *model.StepContext) error {
if !e.config.Enabled {
log.Debug(ctx, "PolicyEnforcer: plugin disabled, skipping")
return nil
}
// Extract action from the message
action := extractAction(ctx.Request.URL.Path, ctx.Body)
if !e.config.IsActionEnabled(action) {
if e.config.DebugLogging {
log.Debugf(ctx, "PolicyEnforcer: action %q not in configured actions %v, skipping", action, e.config.Actions)
}
return nil
}
if e.config.DebugLogging {
log.Debugf(ctx, "PolicyEnforcer: evaluating policies for action %q (modules=%v)", action, e.evaluator.ModuleNames())
}
violations, err := e.evaluator.Evaluate(ctx, ctx.Body)
if err != nil {
// Fail closed: evaluation error → NACK
log.Errorf(ctx, err, "PolicyEnforcer: policy evaluation failed: %v", err)
return model.NewBadReqErr(fmt.Errorf("policy evaluation error: %w", err))
}
if len(violations) == 0 {
if e.config.DebugLogging {
log.Debugf(ctx, "PolicyEnforcer: message compliant for action %q", action)
}
return nil
}
// Non-compliant: NACK with all violation messages
msg := fmt.Sprintf("policy violation(s): %s", strings.Join(violations, "; "))
log.Warnf(ctx, "PolicyEnforcer: %s", msg)
return model.NewBadReqErr(fmt.Errorf("%s", msg))
}
// Close is a no-op for the policy enforcer (no resources to release).
func (e *PolicyEnforcer) Close() {}
// extractAction gets the beckn action from the URL path or message body.
func extractAction(urlPath string, body []byte) string {
// Try URL path first: /bap/receiver/{action} or /bpp/caller/{action}
parts := strings.Split(strings.Trim(urlPath, "/"), "/")
if len(parts) >= 3 {
return parts[len(parts)-1]
}
// Fallback: extract from body context.action
var payload struct {
Context struct {
Action string `json:"action"`
} `json:"context"`
}
if err := json.Unmarshal(body, &payload); err == nil && payload.Context.Action != "" {
return payload.Context.Action
}
return ""
}

520
pkg/plugin/implementation/policyenforcer/enforcer_test.go
View File

@@ -1,520 +0,0 @@
package policyenforcer
import (
"context"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"github.com/beckn-one/beckn-onix/pkg/model"
)
// Helper: create a StepContext with the given action path and JSON body.
func makeStepCtx(action string, body string) *model.StepContext {
req, _ := http.NewRequest("POST", "/bpp/caller/"+action, nil)
return &model.StepContext{
Context: context.Background(),
Request: req,
Body: []byte(body),
}
}
// Helper: write a .rego file to a temp dir and return the dir path.
func writePolicyDir(t *testing.T, filename, content string) string {
t.Helper()
dir := t.TempDir()
err := os.WriteFile(filepath.Join(dir, filename), []byte(content), 0644)
if err != nil {
t.Fatalf("failed to write policy file: %v", err)
}
return dir
}
// --- Config Tests ---
func TestParseConfig_RequiresPolicySource(t *testing.T) {
_, err := ParseConfig(map[string]string{})
if err == nil {
t.Fatal("expected error when no policyPaths given")
}
}
func TestParseConfig_Defaults(t *testing.T) {
cfg, err := ParseConfig(map[string]string{"policyPaths": "/tmp"})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if cfg.Query != "data.policy.violations" {
t.Errorf("expected default query 'data.policy.violations', got %q", cfg.Query)
}
if len(cfg.Actions) != 0 {
t.Errorf("expected empty default actions (all enabled), got %v", cfg.Actions)
}
if !cfg.Enabled {
t.Error("expected enabled=true by default")
}
}
func TestParseConfig_RuntimeConfigForwarding(t *testing.T) {
cfg, err := ParseConfig(map[string]string{
"policyPaths": "/tmp",
"minDeliveryLeadHours": "6",
"customParam": "value",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if cfg.RuntimeConfig["minDeliveryLeadHours"] != "6" {
t.Errorf("expected minDeliveryLeadHours=6, got %q", cfg.RuntimeConfig["minDeliveryLeadHours"])
}
if cfg.RuntimeConfig["customParam"] != "value" {
t.Errorf("expected customParam=value, got %q", cfg.RuntimeConfig["customParam"])
}
}
func TestParseConfig_CustomActions(t *testing.T) {
cfg, err := ParseConfig(map[string]string{
"policyPaths": "/tmp",
"actions": "confirm, select, init",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(cfg.Actions) != 3 {
t.Fatalf("expected 3 actions, got %d: %v", len(cfg.Actions), cfg.Actions)
}
expected := []string{"confirm", "select", "init"}
for i, want := range expected {
if cfg.Actions[i] != want {
t.Errorf("action[%d] = %q, want %q", i, cfg.Actions[i], want)
}
}
}
func TestParseConfig_PolicyPaths(t *testing.T) {
cfg, err := ParseConfig(map[string]string{
"policyPaths": "https://example.com/a.rego, https://example.com/b.rego",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(cfg.PolicyPaths) != 2 {
t.Fatalf("expected 2 paths, got %d: %v", len(cfg.PolicyPaths), cfg.PolicyPaths)
}
if cfg.PolicyPaths[0] != "https://example.com/a.rego" {
t.Errorf("path[0] = %q", cfg.PolicyPaths[0])
}
}
// --- Evaluator Tests (with inline policies) ---
func TestEvaluator_NoViolations(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains msg if {
input.value < 0
msg := "value is negative"
}
`
dir := writePolicyDir(t, "test.rego", policy)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator failed: %v", err)
}
violations, err := eval.Evaluate(context.Background(), []byte(`{"value": 10}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 0 {
t.Errorf("expected 0 violations, got %d: %v", len(violations), violations)
}
}
func TestEvaluator_WithViolation(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains msg if {
input.value < 0
msg := "value is negative"
}
`
dir := writePolicyDir(t, "test.rego", policy)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator failed: %v", err)
}
violations, err := eval.Evaluate(context.Background(), []byte(`{"value": -5}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 1 {
t.Fatalf("expected 1 violation, got %d: %v", len(violations), violations)
}
if violations[0] != "value is negative" {
t.Errorf("unexpected violation: %q", violations[0])
}
}
func TestEvaluator_RuntimeConfig(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains msg if {
input.value > to_number(data.config.maxValue)
msg := "value exceeds maximum"
}
`
dir := writePolicyDir(t, "test.rego", policy)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", map[string]string{"maxValue": "100"})
if err != nil {
t.Fatalf("NewEvaluator failed: %v", err)
}
// Under limit
violations, err := eval.Evaluate(context.Background(), []byte(`{"value": 50}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 0 {
t.Errorf("expected 0 violations for value=50, got %v", violations)
}
// Over limit
violations, err = eval.Evaluate(context.Background(), []byte(`{"value": 150}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 1 {
t.Errorf("expected 1 violation for value=150, got %v", violations)
}
}
func TestEvaluator_SkipsTestFiles(t *testing.T) {
dir := t.TempDir()
policy := `
package policy
import rego.v1
violations contains "always" if { true }
`
os.WriteFile(filepath.Join(dir, "policy.rego"), []byte(policy), 0644)
// Test file would cause compilation issues if loaded (different package)
testFile := `
package policy_test
import rego.v1
import data.policy
test_something if { count(policy.violations) > 0 }
`
os.WriteFile(filepath.Join(dir, "policy_test.rego"), []byte(testFile), 0644)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator should skip _test.rego files, but failed: %v", err)
}
violations, err := eval.Evaluate(context.Background(), []byte(`{}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 1 {
t.Errorf("expected 1 violation, got %d", len(violations))
}
}
func TestEvaluator_InvalidJSON(t *testing.T) {
policy := `
package policy
import rego.v1
violations := set()
`
dir := writePolicyDir(t, "test.rego", policy)
eval, err := NewEvaluator([]string{dir}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator failed: %v", err)
}
_, err = eval.Evaluate(context.Background(), []byte(`not json`))
if err == nil {
t.Error("expected error for invalid JSON")
}
}
// --- Evaluator URL Fetch Tests ---
func TestEvaluator_FetchFromURL(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains msg if {
input.value < 0
msg := "value is negative"
}
`
// Serve the policy via a local HTTP server
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
w.Write([]byte(policy))
}))
defer srv.Close()
eval, err := NewEvaluator([]string{srv.URL + "/test_policy.rego"}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator with URL failed: %v", err)
}
// Compliant
violations, err := eval.Evaluate(context.Background(), []byte(`{"value": 10}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 0 {
t.Errorf("expected 0 violations, got %v", violations)
}
// Non-compliant
violations, err = eval.Evaluate(context.Background(), []byte(`{"value": -1}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 1 {
t.Errorf("expected 1 violation, got %v", violations)
}
}
func TestEvaluator_FetchURL_NotFound(t *testing.T) {
srv := httptest.NewServer(http.NotFoundHandler())
defer srv.Close()
_, err := NewEvaluator([]string{srv.URL + "/missing.rego"}, "data.policy.violations", nil)
if err == nil {
t.Fatal("expected error for 404 URL")
}
}
func TestEvaluator_FetchURL_InvalidScheme(t *testing.T) {
_, err := NewEvaluator([]string{"ftp://example.com/policy.rego"}, "data.policy.violations", nil)
if err == nil {
t.Fatal("expected error for ftp:// scheme")
}
}
func TestEvaluator_MixedLocalAndURL(t *testing.T) {
// Local policy
localPolicy := `
package policy
import rego.v1
violations contains "local_violation" if { input.local_bad }
`
dir := writePolicyDir(t, "local.rego", localPolicy)
// Remote policy (different rule, same package)
remotePolicy := `
package policy
import rego.v1
violations contains "remote_violation" if { input.remote_bad }
`
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(remotePolicy))
}))
defer srv.Close()
eval, err := NewEvaluator([]string{dir, srv.URL + "/remote.rego"}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator failed: %v", err)
}
// Trigger both violations
violations, err := eval.Evaluate(context.Background(), []byte(`{"local_bad": true, "remote_bad": true}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 2 {
t.Errorf("expected 2 violations (local+remote), got %d: %v", len(violations), violations)
}
}
// --- Evaluator with local file path in policySources ---
func TestEvaluator_LocalFilePath(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "from_file" if { input.bad }
`
dir := t.TempDir()
policyPath := filepath.Join(dir, "local_policy.rego")
os.WriteFile(policyPath, []byte(policy), 0644)
eval, err := NewEvaluator([]string{policyPath}, "data.policy.violations", nil)
if err != nil {
t.Fatalf("NewEvaluator with local path failed: %v", err)
}
violations, err := eval.Evaluate(context.Background(), []byte(`{"bad": true}`))
if err != nil {
t.Fatalf("Evaluate failed: %v", err)
}
if len(violations) != 1 || violations[0] != "from_file" {
t.Errorf("expected [from_file], got %v", violations)
}
}
// --- Enforcer Integration Tests ---
func TestEnforcer_Compliant(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "blocked" if { input.context.action == "confirm"; input.block }
`
dir := writePolicyDir(t, "test.rego", policy)
enforcer, err := New(map[string]string{
"policyPaths": dir,
"query": "data.policy.violations",
"actions": "confirm",
})
if err != nil {
t.Fatalf("New failed: %v", err)
}
ctx := makeStepCtx("confirm", `{"context": {"action": "confirm"}, "block": false}`)
err = enforcer.Run(ctx)
if err != nil {
t.Errorf("expected nil error for compliant message, got: %v", err)
}
}
func TestEnforcer_NonCompliant(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "blocked" if { input.context.action == "confirm" }
`
dir := writePolicyDir(t, "test.rego", policy)
enforcer, err := New(map[string]string{
"policyPaths": dir,
"query": "data.policy.violations",
"actions": "confirm",
})
if err != nil {
t.Fatalf("New failed: %v", err)
}
ctx := makeStepCtx("confirm", `{"context": {"action": "confirm"}}`)
err = enforcer.Run(ctx)
if err == nil {
t.Fatal("expected error for non-compliant message, got nil")
}
// Should be a BadReqErr
if _, ok := err.(*model.BadReqErr); !ok {
t.Errorf("expected *model.BadReqErr, got %T: %v", err, err)
}
}
func TestEnforcer_SkipsNonMatchingAction(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "blocked" if { true }
`
dir := writePolicyDir(t, "test.rego", policy)
enforcer, err := New(map[string]string{
"policyPaths": dir,
"query": "data.policy.violations",
"actions": "confirm",
})
if err != nil {
t.Fatalf("New failed: %v", err)
}
// Non-compliant body, but action is "search" — not in configured actions
ctx := makeStepCtx("search", `{"context": {"action": "search"}}`)
err = enforcer.Run(ctx)
if err != nil {
t.Errorf("expected nil for non-matching action, got: %v", err)
}
}
func TestEnforcer_DisabledPlugin(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "blocked" if { true }
`
dir := writePolicyDir(t, "test.rego", policy)
enforcer, err := New(map[string]string{
"policyPaths": dir,
"query": "data.policy.violations",
"enabled": "false",
})
if err != nil {
t.Fatalf("New failed: %v", err)
}
ctx := makeStepCtx("confirm", `{"context": {"action": "confirm"}}`)
err = enforcer.Run(ctx)
if err != nil {
t.Errorf("expected nil for disabled plugin, got: %v", err)
}
}
// --- Enforcer with URL-sourced policy ---
func TestEnforcer_PolicyFromURL(t *testing.T) {
policy := `
package policy
import rego.v1
violations contains "blocked" if { input.context.action == "confirm" }
`
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(policy))
}))
defer srv.Close()
enforcer, err := New(map[string]string{
"policyPaths": srv.URL + "/block_confirm.rego",
"query": "data.policy.violations",
"actions": "confirm",
})
if err != nil {
t.Fatalf("New failed: %v", err)
}
ctx := makeStepCtx("confirm", `{"context": {"action": "confirm"}}`)
err = enforcer.Run(ctx)
if err == nil {
t.Fatal("expected error from URL-sourced policy, got nil")
}
if _, ok := err.(*model.BadReqErr); !ok {
t.Errorf("expected *model.BadReqErr, got %T", err)
}
}
// --- extractAction Tests ---
func TestExtractAction_FromURL(t *testing.T) {
action := extractAction("/bpp/caller/confirm", nil)
if action != "confirm" {
t.Errorf("expected 'confirm', got %q", action)
}
}
func TestExtractAction_FromBody(t *testing.T) {
body := []byte(`{"context": {"action": "select"}}`)
action := extractAction("/x", body)
if action != "select" {
t.Errorf("expected 'select', got %q", action)
}
}

220
pkg/plugin/implementation/policyenforcer/evaluator.go
View File

@@ -1,220 +0,0 @@
package policyenforcer
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"os"
"path"
"path/filepath"
"strings"
"time"
"github.com/open-policy-agent/opa/v1/ast"
"github.com/open-policy-agent/opa/v1/rego"
"github.com/open-policy-agent/opa/v1/storage/inmem"
)
// Evaluator wraps the OPA engine: loads and compiles .rego files at startup,
// then evaluates messages against the compiled policy set.
type Evaluator struct {
preparedQuery rego.PreparedEvalQuery
query string
runtimeConfig map[string]string
moduleNames []string // names of loaded .rego modules
}
// ModuleNames returns the names of the loaded .rego policy modules.
func (e *Evaluator) ModuleNames() []string {
return e.moduleNames
}
// policyFetchTimeout is the HTTP timeout for fetching remote .rego files.
const policyFetchTimeout = 30 * time.Second
// maxPolicySize is the maximum size of a single .rego file fetched from a URL (1 MB).
const maxPolicySize = 1 << 20
// NewEvaluator creates an Evaluator by loading .rego files from local paths
// and/or URLs, then compiling them. runtimeConfig is passed to Rego as data.config.
func NewEvaluator(policyPaths []string, query string, runtimeConfig map[string]string) (*Evaluator, error) {
modules := make(map[string]string)
// Load from policyPaths (each entry auto-detected as URL, directory, or file)
for _, source := range policyPaths {
if isURL(source) {
name, content, err := fetchPolicy(source)
if err != nil {
return nil, fmt.Errorf("failed to fetch policy from %s: %w", source, err)
}
modules[name] = content
} else if info, err := os.Stat(source); err == nil && info.IsDir() {
// Directory — load all .rego files inside
entries, err := os.ReadDir(source)
if err != nil {
return nil, fmt.Errorf("failed to read policy directory %s: %w", source, err)
}
for _, entry := range entries {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".rego") || strings.HasSuffix(entry.Name(), "_test.rego") {
continue
}
fpath := filepath.Join(source, entry.Name())
data, err := os.ReadFile(fpath)
if err != nil {
return nil, fmt.Errorf("failed to read policy file %s: %w", fpath, err)
}
modules[entry.Name()] = string(data)
}
} else {
// Local file path
data, err := os.ReadFile(source)
if err != nil {
return nil, fmt.Errorf("failed to read policy file %s: %w", source, err)
}
modules[filepath.Base(source)] = string(data)
}
}
if len(modules) == 0 {
return nil, fmt.Errorf("no .rego policy files found from any configured source")
}
// Compile modules to catch syntax errors early
compiler, err := ast.CompileModulesWithOpt(modules, ast.CompileOpts{ParserOptions: ast.ParserOptions{RegoVersion: ast.RegoV1}})
if err != nil {
return nil, fmt.Errorf("failed to compile rego modules: %w", err)
}
// Build data.config from runtime config
store := map[string]interface{}{
"config": toInterfaceMap(runtimeConfig),
}
pq, err := rego.New(
rego.Query(query),
rego.Compiler(compiler),
rego.Store(inmem.NewFromObject(store)),
).PrepareForEval(context.Background())
if err != nil {
return nil, fmt.Errorf("failed to prepare rego query %q: %w", query, err)
}
names := make([]string, 0, len(modules))
for name := range modules {
names = append(names, name)
}
return &Evaluator{
preparedQuery: pq,
query: query,
runtimeConfig: runtimeConfig,
moduleNames: names,
}, nil
}
// isURL checks if a source string looks like a remote URL.
func isURL(source string) bool {
return strings.HasPrefix(source, "http://") || strings.HasPrefix(source, "https://")
}
// fetchPolicy downloads a .rego file from a URL and returns (filename, content, error).
func fetchPolicy(rawURL string) (string, string, error) {
parsed, err := url.Parse(rawURL)
if err != nil {
return "", "", fmt.Errorf("invalid URL: %w", err)
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return "", "", fmt.Errorf("unsupported URL scheme %q (only http and https are supported)", parsed.Scheme)
}
client := &http.Client{Timeout: policyFetchTimeout}
resp, err := client.Get(rawURL)
if err != nil {
return "", "", fmt.Errorf("HTTP request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", "", fmt.Errorf("HTTP %d from %s", resp.StatusCode, rawURL)
}
// Read with size limit
limited := io.LimitReader(resp.Body, maxPolicySize+1)
body, err := io.ReadAll(limited)
if err != nil {
return "", "", fmt.Errorf("failed to read response body: %w", err)
}
if len(body) > maxPolicySize {
return "", "", fmt.Errorf("policy file exceeds maximum size of %d bytes", maxPolicySize)
}
// Derive filename from URL path
name := path.Base(parsed.Path)
if name == "" || name == "." || name == "/" {
name = "policy.rego"
}
if !strings.HasSuffix(name, ".rego") {
name += ".rego"
}
return name, string(body), nil
}
// Evaluate runs the compiled policy against a JSON message body.
// Returns a list of violation strings (empty = compliant).
func (e *Evaluator) Evaluate(ctx context.Context, body []byte) ([]string, error) {
var input interface{}
if err := json.Unmarshal(body, &input); err != nil {
return nil, fmt.Errorf("failed to parse message body as JSON: %w", err)
}
rs, err := e.preparedQuery.Eval(ctx, rego.EvalInput(input))
if err != nil {
return nil, fmt.Errorf("rego evaluation failed: %w", err)
}
return extractViolations(rs)
}
// extractViolations pulls string violations from the OPA result set.
// The query is expected to return a set of strings.
func extractViolations(rs rego.ResultSet) ([]string, error) {
if len(rs) == 0 {
return nil, nil
}
var violations []string
for _, result := range rs {
for _, expr := range result.Expressions {
switch v := expr.Value.(type) {
case []interface{}:
// Result is a list (from set)
for _, item := range v {
if s, ok := item.(string); ok {
violations = append(violations, s)
}
}
case map[string]interface{}:
// OPA sometimes returns sets as maps with string keys
for key := range v {
violations = append(violations, key)
}
}
}
}
return violations, nil
}
// toInterfaceMap converts map[string]string to map[string]interface{} for OPA store.
func toInterfaceMap(m map[string]string) map[string]interface{} {
result := make(map[string]interface{}, len(m))
for k, v := range m {
result[k] = v
}
return result
}

10
pkg/plugin/manager.go
View File

@@ -257,21 +257,21 @@ func (m *Manager) Step(ctx context.Context, cfg *Config) (definition.Step, error
return step, error
}
// PolicyEnforcer returns a PolicyEnforcer instance based on the provided configuration.
// PolicyChecker returns a PolicyChecker instance based on the provided configuration.
// It registers a cleanup function for resource management.
func (m *Manager) PolicyEnforcer(ctx context.Context, cfg *Config) (definition.PolicyEnforcer, error) {
pp, err := provider[definition.PolicyEnforcerProvider](m.plugins, cfg.ID)
func (m *Manager) PolicyChecker(ctx context.Context, cfg *Config) (definition.PolicyChecker, error) {
pp, err := provider[definition.PolicyCheckerProvider](m.plugins, cfg.ID)
if err != nil {
return nil, fmt.Errorf("failed to load provider for %s: %w", cfg.ID, err)
}
enforcer, closer, err := pp.New(ctx, cfg.Config)
checker, closer, err := pp.New(ctx, cfg.Config)
if err != nil {
return nil, err
}
if closer != nil {
m.closers = append(m.closers, closer)
}
return enforcer, nil
return checker, nil
}
// Cache returns a Cache instance based on the provided configuration.